package com.rxjava;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

//join：如果一个Observable发射了一条数据，只要在另一个Observable发射的数据定义的时间窗口内，就结合两个Observable发射的数据，然后发射结合后的数据。
//目标Observable和源Observable发射的数据都有一个有效时间限制，比如目标发射了一条数据（a）有效期为3s，过了2s后，源发射了一条数据（b），因为2s<3s，目标的那条数据还在有效期，所以可以组合为ab；
//再过2s，源又发射了一条数据（c）,这时候一共过去了4s，目标的数据a已经过期，所以不能组合了…
public class JoinDemo {

	public static void main(String[] args) {
		Observable<Integer> obs1 = Observable.create(new ObservableOnSubscribe<Integer>() {
			@Override
			public void subscribe(ObservableEmitter<Integer> emitter) {
				for (int i = 0; i < 5; i++) {
					emitter.onNext(i);
					try {
						Thread.sleep(1000);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}
		});
		//join
		Observable.just("srcObs-")
		.join(obs1,
	        //接受从源Observable发射来的数据，并返回一个Observable，
	        //这个Observable的生命周期决定了源Observable发射出来数据的有效期
	        new Function<String, Observable<Long>>() {
	            @Override
	            public Observable<Long> apply(String s) {
	                return Observable.timer(3000, TimeUnit.MILLISECONDS);
	            }
	        },
	        //接受从目标Observable发射来的数据，并返回一个Observable，
	        //这个Observable的生命周期决定了目标Observable发射出来数据的有效期
	        new Function<Integer, Observable<Long>>() {
	            @Override
	            public Observable<Long> apply(Integer integer) {
	                return Observable.timer(2000, TimeUnit.MILLISECONDS);
	            }
	        },
	        //接收从源Observable和目标Observable发射来的数据，并返回最终组合完的数据
	        new BiFunction<String, Integer, String>() {
	            @Override
	            public String apply(String str1, Integer integer) {
	                return str1 + integer;
	            }
	        }
		)
		.subscribe(x -> System.out.println(x));

	}
}
